package driver

import (
	"context"
	"fmt"
	"os"
	"sync/atomic"
	"testing"
	"time"

	"github.com/datazip-inc/olake/pkg/waljs"
	"github.com/jmoiron/sqlx"
)

// TEMPORARY repro file (delete after fix): PreCDC seeds state from the WAL write pointer
// while pg_replication_slot_advance clamps to flush — both ladder tests hunt the trigger load.
const reproSlotName = "olake_repro_slot"

// setupReproInfra connects, recreates the repro slot and load table, and registers cleanup.
func setupReproInfra(ctx context.Context, t *testing.T) *sqlx.DB {
	t.Helper()
	dsn := os.Getenv("OLAKE_PG_REPRO_DSN")
	if dsn == "" {
		t.Skip("set OLAKE_PG_REPRO_DSN (compose default: postgres://postgres:secret1234@localhost:5433/postgres?sslmode=disable)")
	}

	db, err := sqlx.ConnectContext(ctx, "pgx", dsn)
	if err != nil {
		t.Fatalf("connect: %s", err)
	}
	t.Cleanup(func() { db.Close() })

	_, _ = db.ExecContext(ctx, fmt.Sprintf("SELECT pg_drop_replication_slot('%s')", reproSlotName))
	if _, err := db.ExecContext(ctx, fmt.Sprintf("SELECT pg_create_logical_replication_slot('%s', 'pgoutput')", reproSlotName)); err != nil {
		t.Fatalf("create slot: %s", err)
	}
	t.Cleanup(func() {
		_, _ = db.ExecContext(context.Background(), fmt.Sprintf("SELECT pg_drop_replication_slot('%s')", reproSlotName))
	})

	if _, err := db.ExecContext(ctx, "CREATE TABLE IF NOT EXISTS olake_repro_load (id bigserial PRIMARY KEY, payload text)"); err != nil {
		t.Fatalf("create load table: %s", err)
	}
	t.Cleanup(func() {
		_, _ = db.ExecContext(context.Background(), "DROP TABLE IF EXISTS olake_repro_load")
	})
	return db
}

// runBootstrapAttempts replays the PreCDC bootstrap plus the next sync's validation until
// deadline; on divergence it fails the test with full forensics, headlined by describe().
func runBootstrapAttempts(ctx context.Context, t *testing.T, db *sqlx.DB, deadline time.Time, describe func() string) (attempts, writeAhead int) {
	t.Helper()
	for time.Now().Before(deadline) {
		attempts++

		var writePtr, flushPtr string
		var gapOpen bool
		if err := db.QueryRowContext(ctx, "SELECT w::text, f::text, w > f FROM (SELECT pg_current_wal_lsn() w, pg_current_wal_flush_lsn() f) s").Scan(&writePtr, &flushPtr, &gapOpen); err != nil {
			t.Fatalf("attempt %d: read wal pointers: %s", attempts, err)
		}
		if gapOpen {
			writeAhead++
		}

		// first sync bootstrap, verbatim from PreCDC (cdc.go:52-56)
		slot, err := waljs.GetSlotPosition(ctx, db, reproSlotName)
		if err != nil {
			t.Fatalf("attempt %d: get slot position: %s", attempts, err)
		}
		state := waljs.WALState{LSN: slot.CurrentLSN.String()}
		if err := waljs.AdvanceLSN(ctx, db, reproSlotName, slot.CurrentLSN.String()); err != nil {
			t.Fatalf("attempt %d: advance: %s", attempts, err)
		}

		// next sync: StreamChanges re-reads the slot and validates (cdc.go:133,152)
		after, err := waljs.GetSlotPosition(ctx, db, reproSlotName)
		if err != nil {
			t.Fatalf("attempt %d: re-read slot: %s", attempts, err)
		}
		if verr := validateGlobalState(state, after.LSN); verr != nil {
			// hypothesis part 2: the end_lsn AdvanceLSN discards reports where the slot really stopped
			var name, endLSN, confirmed string
			requested := after.CurrentLSN.String()
			if err := db.QueryRowContext(ctx, fmt.Sprintf(waljs.AdvanceLSNTemplate, reproSlotName, requested)).Scan(&name, &endLSN); err != nil {
				endLSN = "scan failed: " + err.Error()
			}
			if err := db.QueryRowContext(ctx, fmt.Sprintf("SELECT confirmed_flush_lsn FROM pg_replication_slots WHERE slot_name = '%s'", reproSlotName)).Scan(&confirmed); err != nil {
				confirmed = "scan failed: " + err.Error()
			}
			t.Fatalf("REPRODUCED at %s\n"+
				"  attempt %d of this level; %d of %d reads saw write ahead of flush\n"+
				"  wal pointers at bootstrap: write=%s flush=%s\n"+
				"  state seeded from pg_current_wal_lsn: %s\n"+
				"  slot confirmed_flush after advance:   %s (%d bytes short)\n"+
				"  next sync fails with: %s\n"+
				"  follow-up advance requested=%s returned end_lsn=%s, slot confirmed_flush=%s (end_lsn is the correct seed)",
				describe(), attempts, writeAhead, attempts, writePtr, flushPtr,
				state.LSN, after.LSN, uint64(slot.CurrentLSN-after.LSN), verr,
				requested, endLSN, confirmed)
		}
	}
	return attempts, writeAhead
}

// Default synchronous_commit: ramps rows per bulk statement — the window only opens once one
// statement's WAL exceeds wal_buffers (4 MB stock), so the ladder brackets that boundary.
func TestReproPreCDCBootstrapLSNDivergence(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()
	db := setupReproInfra(ctx, t)

	loadConn, err := db.Connx(ctx)
	if err != nil {
		t.Fatalf("load conn: %s", err)
	}
	var rowsPerStmt, rowsWritten atomic.Int64
	loadCtx, stopLoad := context.WithCancel(ctx)
	loadDone := make(chan struct{})
	go func() {
		defer close(loadDone)
		for i := 0; loadCtx.Err() == nil; i++ {
			if i%3 == 0 {
				_, _ = loadConn.ExecContext(loadCtx, "TRUNCATE olake_repro_load")
			}
			n := rowsPerStmt.Load()
			if _, err := loadConn.ExecContext(loadCtx, fmt.Sprintf("INSERT INTO olake_repro_load (payload) SELECT repeat('x', 1000) FROM generate_series(1, %d)", n)); err == nil {
				rowsWritten.Add(n)
			}
		}
	}()
	defer func() { stopLoad(); <-loadDone; loadConn.Close() }()

	levels := []int64{500, 1000, 2000, 4000, 8000, 16000, 32000}
	rowsPerStmt.Store(levels[0])
	for _, n := range levels {
		rowsPerStmt.Store(n)
		rowsWritten.Store(0)
		levelStart := time.Now()
		describe := func() string {
			rps := float64(rowsWritten.Load()) / time.Since(levelStart).Seconds()
			return fmt.Sprintf("sync-commit level %d rows/stmt (~%.1f MB WAL/stmt, ~%.0f rows/s achieved)", n, float64(n)/1000.0, rps)
		}
		attempts, writeAhead := runBootstrapAttempts(ctx, t, db, levelStart.Add(20*time.Second), describe)
		t.Logf("%s: %d bootstrap attempts, %d saw write ahead of flush — invariant held", describe(), attempts, writeAhead)
	}
	t.Logf("no sync-commit load level reproduced the divergence; invariant held at every level")
}

// Async-commit ingest (SET synchronous_commit = off, a documented knob for bulk/event pipelines):
// ramps the rate of small statements — commits stop closing the window, only the walwriter is left.
func TestReproPreCDCBootstrapLSNDivergenceAsyncCommit(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()
	db := setupReproInfra(ctx, t)

	loadConn, err := db.Connx(ctx)
	if err != nil {
		t.Fatalf("load conn: %s", err)
	}
	if _, err := loadConn.ExecContext(ctx, "SET synchronous_commit = off"); err != nil {
		t.Fatalf("set async commit: %s", err)
	}

	// ~100 KB per statement, far below wal_buffers on purpose: any window at these levels
	// comes from async commits piling up, not from single-statement buffer pressure
	const rowsEach = 100
	var intervalNS, stmtsDone atomic.Int64
	loadCtx, stopLoad := context.WithCancel(ctx)
	loadDone := make(chan struct{})
	go func() {
		defer close(loadDone)
		for i := 0; loadCtx.Err() == nil; i++ {
			if i%100 == 0 {
				_, _ = loadConn.ExecContext(loadCtx, "TRUNCATE olake_repro_load")
			}
			if _, err := loadConn.ExecContext(loadCtx, fmt.Sprintf("INSERT INTO olake_repro_load (payload) SELECT repeat('x', 1000) FROM generate_series(1, %d)", rowsEach)); err == nil {
				stmtsDone.Add(1)
			}
			if d := intervalNS.Load(); d > 0 {
				select {
				case <-loadCtx.Done():
					return
				case <-time.After(time.Duration(d)):
				}
			}
		}
	}()
	defer func() { stopLoad(); <-loadDone; loadConn.Close() }()

	// target statements/sec; 0 = unthrottled back-to-back
	levels := []int64{5, 10, 25, 50, 100, 250, 0}
	for _, rate := range levels {
		if rate > 0 {
			intervalNS.Store(int64(time.Second) / rate)
		} else {
			intervalNS.Store(0)
		}
		stmtsDone.Store(0)
		levelStart := time.Now()
		target := "max (unthrottled)"
		if rate > 0 {
			target = fmt.Sprintf("%d stmt/s", rate)
		}
		describe := func() string {
			s := float64(stmtsDone.Load()) / time.Since(levelStart).Seconds()
			return fmt.Sprintf("async-commit level target %s × %d rows/stmt (achieved ~%.0f stmt/s ≈ %.1f MB/s WAL)", target, rowsEach, s, s*float64(rowsEach)/1000.0)
		}
		attempts, writeAhead := runBootstrapAttempts(ctx, t, db, levelStart.Add(20*time.Second), describe)
		t.Logf("%s: %d bootstrap attempts, %d saw write ahead of flush — invariant held", describe(), attempts, writeAhead)
	}
	t.Logf("no async-commit rate reproduced the divergence; invariant held at every level")
}